package cn.itcast.flink.window.time;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
/**
* 基于事件时间EventTime滚动窗口统计：每隔5s，计算5秒内，每个用户的订单总金额
* TODO： 添加 Watermaker 来解决一定程度上的数据延迟和数据乱序问题
* TODO: 使用 OutputTag+allowedLateness，解决数据丢失问题
*/
public class StreamEventTimeLatenessWindow_5 {
public static void main(String[] args) throws Exception {
                // 1. 执行环境-env
                StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                env.setParallelism(1) ;
                // step1. 设置使用事件时间EventTime进行窗口计算
                env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
                // 2. 数据源-source
                DataStreamSource<Order> orderStream = env.addSource(new OrderSource());
                // step2. 设置事件时间字段和添加Watermark
                FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss:SSS");
                SingleOutputStreamOperator<Order> timeStream = orderStream.assignTimestampsAndWatermarks(
                new AssignerWithPeriodicWatermarks<Order>() {
                // Watermark延迟时间（允许最大数据乱序时间）, 此处设置最大延迟3秒
                Long maxOutOfOrderness = 2000L;
                // 当前最大事件时间
                Long currentMaxTimestamp = Long.MIN_VALUE + 2000L;
                // 最新的水位时间
                Long lastEmittedWatermark = Long.MIN_VALUE;
                // 获取当前的watermark
                @Nullable
                @Override
                public Watermark getCurrentWatermark() {
                // 依据当前数据中事件时间，计算出水位Watermark值
                long potentialWatermark = currentMaxTimestamp - maxOutOfOrderness;
                // 比较当前数据计算Watermark值与上次数据Watermark值大小，设置Watermark
                if(potentialWatermark >= lastEmittedWatermark){
                lastEmittedWatermark = potentialWatermark ; }
                return new Watermark(lastEmittedWatermark);
                }
                // 获取当前的时间戳
                @Override
                public long extractTimestamp(Order order, long previousElementTimestamp) {
                Long eventTime = order.getOrderTime();
                // 比较当前数据事件时间与最大事件时间大小
                if (eventTime > currentMaxTimestamp) {
                currentMaxTimestamp = eventTime;
                }
                // 打印当前数据Watermark值
                System.err.println(
                "key: " + order.getUserId()
                + ", money: " + order.getMoney()
                + ", eventTime: " + format.format(eventTime)
                + ", watermark: " + format.format(getCurrentWatermark().getTimestamp())
                );
                return eventTime;
                } }
                );
                // 3. 数据转换-transformation
                // TODO: 其一、定义一个侧边输出流，用来接收延迟较为严重的数据
                OutputTag<Order> delayTag = new OutputTag<Order>("delayData"){} ;
                // step3. 每隔5s，计算5秒内，每个用户的订单总金额
                SingleOutputStreamOperator<String> resultStream = timeStream
                // 按照用户ID分组
                .keyBy("userId")
                // 窗口大小5秒，基于事件时间EventTime：orderTime
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // TODO: 其二、设置延迟时间大于多少的时候进入侧边输出流
                .allowedLateness(Time.seconds(2))
                // TODO: 其三、设置要进入到哪个侧边输出流
                .sideOutputLateData(delayTag)
                // 窗口聚合操作
                .apply(new WindowFunction<Order, String, Tuple, TimeWindow>() { @Override
public void apply(Tuple tuple,
            TimeWindow window,
            Iterable<Order> input,
            Collector<String> out) throws Exception {
            // 获取窗口StartTime和EndTime
            String startTime = format.format(window.getStart());
            String endTime = format.format(window.getEnd()) ;
            // 获取组合所有订单时间
            List<String> list = new ArrayList<>();
            Integer orderSum = 0 ;
            for (Order order: input){
            list.add(format.format(order.getOrderTime())) ;
            orderSum += order.getMoney() ; }
            String output = "窗口计算结果>>> key: " + tuple.toString()
            + ", startTime: " + startTime + ", endTime: " + endTime
            + ", sum： " + orderSum + ", Orders: " + list;
            out.collect(output);
            }
            });
            // 4. 数据终端-sink
            resultStream.printToErr();
            // TODO: 其四、获取延迟的数据
            DataStream<Order> delayDataStream = resultStream.getSideOutput(delayTag);
            delayDataStream.print("延迟订单数据>>>");
            // 5. 应用执行
            env.execute(StreamEventTimeLatenessWindow_5.class.getSimpleName()) ;
} }